package com.lagou.rpc.consumer.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import lombok.Data;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 */
@Component
@Data
public class RpcClientProxy {

    //服务端上线后所在zookeeper的根节点
    private String rootNote = "/rpcServer";
    //zookeeper 会话对象，启动类传进来
    private ZkClient zkClientSession;
    //根据规则被调用的服务端IP，与Controller层互动需要
    private String choseServer;
    //调用已上线的服务端的rpcClient对象储存容器
    Map<String,RpcClient> rpcClientMap= new HashMap<>();
    //已上线服务端IP及端口号储存容器
    Map<String,Integer> rpcServerMap= new HashMap<>();
    //记录最后一次请求耗时
    Map<String,Long> serverTimeMap= new HashMap<>();
    //记录最后一次请求时的系统时间
    Map<String,String> serverSystemTimeMap= new HashMap<>();

    //Controller层调用，以在浏览器实时显示每台服务器最后的响应耗时情况
    public Map<String, Long> getServerTimeMap() {
        return serverTimeMap;
    }

    public String getChoseServer() {
        return choseServer;
    }

    public void setServerTimeMap(Map<String, Long> serverTimeMap) {
        this.serverTimeMap = serverTimeMap;
    }

    //此方法由启动类中监听zookeeper的服务端子节点列表变化时调用，进行客户端初始化工作
    public List<String> setRpcClient(ZkClient zkClient,List<String> serverS) throws Exception {
        //该类获取zookeeper连接会话对象。
        this.zkClientSession = zkClient;
        //在控制台输出客户端初始化情况及状态信息
        List<String> statusString = new ArrayList<>();
        List<String> serverIps = new ArrayList<>();
        //当没有一个服务端上线的情况下，停止所有初始化工作，结束该方法，继续监听等待服务端上线
        if (serverS.size() == 0){
            statusString.add("暂时无任何在线服务端可连接，客户端持续等待中");
            statusString=serverDown(serverIps, statusString, zkClient);
            return statusString;
        }

        for (String server : serverS) {
            //根据监听到的zookeeper上服务端子节点列表，读取子节点列表上的内容
            Object noteInfo = zkClient.readData(rootNote + "/" + server);
            //纯手工new一个rpcClient
            RpcClient rpcClient = new RpcClient();
            if (noteInfo != null) {
                //zookeeper上服务端子节点名字就是服务端的IP地址，子节点上的内容格式为：IP：端口号
                String[] split = noteInfo.toString().split(":");
                //将获取的服务端IP存到serverIps集合中
                serverIps.add(split[0]);
                if (!rpcServerMap.containsKey(split[0])) {
                    //将本次循环中的服务端IP地址做为Key，端口号做为Value，封装到rpcServerMap容器
                    rpcServerMap.put(split[0], Integer.valueOf(split[1]));
                    //编程题一：初始化时，客户端连接所有在线服务端
                    //rpcClient.initClient(split[0], Integer.valueOf(split[1]));
                    //将本次循环中的服务端IP地址做为Key，纯手工new的rpcClient对象做为Value，封装到rpcClientMap容器
                    rpcClientMap.put(split[0], rpcClient);

                    if (!zkClient.exists("/" + split[0])){
                        String dateTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Calendar.getInstance().getTime());
                        serverSystemTimeMap.put(split[0],dateTime);
                        serverTimeMap.put(split[0],0l);
                        zkClientSession.createEphemeral("/" + split[0],dateTime + "|" + 0);
                    }
                    //客户端控制台状态通知
                    statusString.add("服务端：" + split[0] + "新上线！");
                }else{
                    //客户端控制台状态通知
                    statusString.add("服务端：" + split[0] + "正在持续为您提供全套服务！");
                }
            }
        }
        //-----------------客户端初始化工作继续^_^---------------------------------------------------
        //提示已经下线的服务端
        statusString = serverDown(serverIps, statusString, zkClient);
        //---------客户端初始化工作结束，开始心跳 ^_^------------------
        //客户端每隔5秒周期性的向zookeeper上报目前在线的服务端响应速度及请求服务端时的时间
        heartbeat(zkClient,serverS);
        //放回控制台要显示的状态信息
        return statusString;
    }

    private List<String> serverDown(List<String> serverIps,List<String> statusString,ZkClient zkClient){
        String serverIP ;
        Iterator<Map.Entry<String, Integer>> it = rpcServerMap.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<String, Integer> entry = it.next();
            serverIP = entry.getKey();
            if (!serverIps.contains(serverIP)) {
                statusString.add("服务端：" + serverIP + "已下线！");
                //关闭对应的rpcClient资源
                //rpcClientMap.get(entry.getKey()).close();
                statusString.add("客户端已经与" + serverIP + "服务端断开连接并关闭自身资源");
                zkClient.delete("/" +serverIP);
                serverSystemTimeMap.remove(serverIP);
                serverTimeMap.remove(serverIP);
                //从rpcClientMap容器中删除与下线服务端IP对应的纯手工new的rpcClient对象。
                rpcClientMap.remove(serverIP);
                //从rpcServerMap容器中删除下线服务端IP及端口号
                it.remove();
            }
        }
        return statusString;
    }
    //客户端每隔5秒向zookeeper进行上报
    public void heartbeat(ZkClient zkClient,List<String> serverS){
        //创建定时线程连接池，最大线程数3
        ExecutorService service = Executors.newScheduledThreadPool(3);
        ((ScheduledExecutorService) service).scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                //遍历rpcClientMap容器，向zookeeper相关节点上报每台在线服务端情况
                for (Map.Entry<String, RpcClient> entry : rpcClientMap.entrySet()) {
                    //如果前端从未请求过一次，zookeeper相关节点未创建，那就啥也不报，只报提示！
                    if (!zkClient.exists("/" + entry.getKey())) {
                        System.out.println("注意：" + entry.getKey() + "该服务端不在线，上报节点尚未创建，此次该节点的定时上报更新失败！");
                        if (rpcServerMap.isEmpty()){
                            break;
                        }
                    } else {
                        //更新相关在线服务端在zookeeper上的上报节点，节点位于zookeeper的根目录下，节点名就是服务端IP地址，
                        // 节点内容格式为：请求时间|响应时间
                        zkClient.writeData("/" + entry.getKey(), serverSystemTimeMap.get(entry.getKey()) + "|" + serverTimeMap.get(entry.getKey()));
                        //获取响应时间与当前上报时间
                        Object o = zkClient.readData("/" + entry.getKey());
                        String[] split = o.toString().split("\\|");
                        String dateTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Calendar.getInstance().getTime());
                        SimpleDateFormat format  = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        try {
                            Date parse = format.parse(split[0]);
                            Date parse1 = format.parse(dateTime);
                            //如果当前上报时间大于响应时间20秒以上，判断为超时，即代表该服务端已有超过10秒未收到任何请求。
                            if (parse1.getTime() - parse.getTime() >= 10000){
                                //在超时状态下，zookeeper上的该上报节点响应时间清0，上报时间更新到当前系统时间
                                //响应时间清零即代表，下一次前端请求时有两种可能：一是所有服务端都清零，客户端随机调用服务端
                                //二是0最小，下一次前端请求，直接调用该服务端。
                                zkClient.writeData("/" + entry.getKey(), dateTime + "|" + 0);
                                //serverSystemTimeMap容器与该服务端IP对应的请求时间更新为当前系统时间
                                serverSystemTimeMap.put(entry.getKey(),dateTime);
                                //serverTimeMap容器与该服务端IP对应的响应时间清零
                                serverTimeMap.put(entry.getKey(),0l);
                                //客户端控制台打印状态
                                System.out.println("你好，" + entry.getKey() + "服务端已超过10秒没有收到新的请求，该服务端很寂寞，响应耗时已经清零！");
                            }
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }

                    }

                }
            }
        },0,5,TimeUnit.SECONDS);//三个参数，表示头次不延时立即执行，之后每5秒执行一次
    }


    //此方法由前端请求时，Controller层调用。其目的是根据规则选择合适的服务端进行调用
    public Object choseAndStartSever(Class serviceClass) throws InterruptedException {
        if (rpcClientMap.isEmpty() || rpcServerMap.isEmpty()){
            return null;
        }
        Object proxy = null;
        List<String> serverIps = new ArrayList<>();

        //前端调用未走启动类的监听，因此此处从容器中取出当前在线服务端列表
        for (Map.Entry<String, Integer> entry : rpcServerMap.entrySet()) {
            serverIps.add(entry.getKey());
        }
        //将在线服务端列表做为参数调用choseSever方法根据规则选择要调用的服务端。
        // 这里返回的是可以被调用的服务端IP
        choseServer = choseSever(serverIps);
        //用返回的可被调用的服务端IP从rpcClientMap容器中取出对应的，我们纯手工new的rpcClient
        RpcClient rpcClient = rpcClientMap.get(choseServer);
        //客户端开始连接对应的服务端，并生成代理对象
        proxy = createProxy(serviceClass, rpcClient,choseServer,rpcServerMap.get(choseServer));
        //serverSystemTimeMap容器存储对应服务端IP的当前系统时间
        String dateTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Calendar.getInstance().getTime());
        serverSystemTimeMap.put(choseServer,dateTime);
        //如果zookeeper上没有该被调用服务端的上报节点，那就创一个！
        //如果有节点，就不管了，反正还有定时上报任务。。。。
        if (!zkClientSession.exists("/" + choseServer)){
            zkClientSession.createEphemeral("/" + choseServer,serverSystemTimeMap.get(choseServer) + "|" + serverTimeMap.get(choseServer));
        }
        //生成的接口的实现类代理对象，返回给Controller层
        return proxy;
    }
    //根据规则选择合适的服务端，并返回该服务端IP
    private String choseSever(List<String> serverIps){
        int size = serverIps.size();
        switch (size) {
            case 2 :
                //由于项目要求总共只有两个服务端，因此这里简化判断，判断zookeeper上这两个服务端的上报节点是否都存在
                if (zkClientSession.exists("/" + serverIps.get(0)) && zkClientSession.exists("/" + serverIps.get(1))) {
                    //分别获取zookeeper上这两个服务端的最后响应耗时
                    String time0 = zkClientSession.readData("/" + serverIps.get(0)).toString().split("\\|")[1];
                    String time1 = zkClientSession.readData("/" + serverIps.get(1)).toString().split("\\|")[1];
                    //比对这两个服务端的最后耗时，比对规则是：响应耗时短的被调用，即返回节点上最后一次记录的响应耗时短的服务端IP
                    long cont = Long.valueOf(time0) - Long.valueOf(time1);
                    if (cont == 0) {
                        return serverIps.get(new Random().nextInt(serverIps.size()));
                    } else {
                        if (cont < 0) {
                            return serverIps.get(0);
                        } else {
                            return serverIps.get(1);
                        }
                    }
                    //如果zookeeper上有任意一台服务端上报节点未被创建，即有任意一台服务端响应耗时无法被获取，那么就随机吧^……^!
                } else {
                    return serverIps.get(new Random().nextInt(serverIps.size()));
                }
            default:
                return serverIps.get(new Random().nextInt(serverIps.size()));
        }
    }

    //RpcClient根绝规则连接在线的rpcServer,并生成代理对象
    public Object createProxy(Class serviceClass,RpcClient rpcClient,String ip,Integer port) throws InterruptedException {
        //通过当前线程获取类加载器
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        //1.封装request请求对象
                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(UUID.randomUUID().toString());
                        //通过类的方法获得类名
                        rpcRequest.setClassName(method.getDeclaringClass().getName());
                        //获取方法名
                        rpcRequest.setMethodName(method.getName());
                        //通过方法获取方法参数类型
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        //获取参数信息
                        rpcRequest.setParameters(args);
                        //每次请求时，客户端重新连接服务端
                        rpcClient.initClient(ip, port);
                        try {
                            //3.发送消息
                            Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                            RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                            if (rpcResponse.getError() != null) {
                                throw new RuntimeException(rpcResponse.getError());
                            }
                            //4.返回结果
                            Object result = rpcResponse.getResult();
                            //method.getReturnType()代表方法返回值类型
                            return JSON.parseObject(result.toString(), method.getReturnType());
                        } catch (Exception e) {
                            throw e;
                        } finally {
                            rpcClient.close();
                        }

                    }
                });
    }
}
